package com.hujf.flink.mapreduce;

import com.hujf.flink.DataStreamAPI.Collection;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @Author hujf
 * @Date 2022/5/23 9:39
 * @Description TODO
 */
public class Reduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String,Integer>> source = environment.fromElements(
                new Tuple2<>("A",1),
                new Tuple2<>("B",2),
                new Tuple2<>("C",1),new Tuple2<>("C",1),
                new Tuple2<>("B",1),
                new Tuple2<>("A",1)
        );



        SingleOutputStreamOperator<Tuple2<String, Integer>> reduce = source.keyBy(0)
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                        return new Tuple2<>(value1.f0,value1.f1+value2.f1);
                    }
                });
        reduce.print();
        environment.execute();
    }
}
